/*******************************************************************************
 * This file is part of Crunch Network.
 *
 * Crunch Network is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free Software
 * Foundation, either version 3 of the License, or (at your option) any later version.
 *
 * Crunch Network is distributed in the hope that it will be useful, but WITHOUT ANY
 * WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A
 * PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License along with
 * Crunch Network.  If not, see <http://www.gnu.org/licenses/>.
 ******************************************************************************/

package com.crunch.network;

import java.util.ArrayDeque;
import java.util.HashMap;
import java.util.Queue;

/**
 * {@code MessageDependencyGraph} keeps track of dependencies between messages sent across the network.
 *
 * The implementation is that of a directed acyclic graph. Each message added to the graph is represented by a node.
 * If message B depends on message A, there is an edge pointing from node A to node B. Once a message has no
 * dependencies (incoming edges), it is executed and all outgoing edges from that message node are removed.
 *
 * After a certain amount of time (specified upon construction), messages "expire" and are removed from the graph.
 * This is so that message IDs can be eventually reused, and so that the graph doesn't grow indefinitely large as
 * time goes on. If an unexecuted message expires, an error is triggered, as this indicates a network problem.
 */
abstract class MessageDependencyGraph {
	/**
	 * Initializes the graph.
	 *
	 * @param   messageExpirationTimeMs the amount of time in milliseconds before messages "expire". A message expires
	 *          when it has been in the graph for a long time but has not been executed. Expired messages suggest
	 *          network issues.
	 */
	public MessageDependencyGraph(int messageExpirationTimeMs) {
		graphError = false;

		this.messageExpirationTimeMs = messageExpirationTimeMs;
		messages = new HashMap<Integer, Message>();
		messageExpirationQueue = new ArrayDeque<ExpirationTime>();
	}

	/**
	 * Called when a message should be executed. This method should be overridden by subclasses to perform the
	 * necessary action.
	 *
	 * @param   messageId the ID of the message which should be executed.
	 */
	public abstract void executeMessage(int messageId);

	/**
	 * Adds a message to the graph. Initially the new message is not marked as "ready to execute".
	 *
	 * @param   messageId the ID of the message to add.
	 * @param   dependencyIds the list of IDs for messages which this message is dependent on. If these messages have
	 *          not yet been added to the graph, placeholder nodes will be automatically created. If {@code null}, the
	 *          message is considered to have no dependencies.
	 * @param   currentTimeMs the current time in milliseconds.
	 * @throws  NetworkException if the graph has been previously invalidated, the message has already been added, or a
	 *          cycle is detected. Only "trivial" cycles are detected here, but all cycles will eventually be caught
	 *          within the message expiration time limit.
	 */
	public final void addMessage(int messageId, int[] dependencyIds, long currentTimeMs)
			throws NetworkException {
		if (graphError) {
			throw new NetworkException("Message dependency graph is invalid");
		}

		int depLength = (dependencyIds == null) ? 0 : dependencyIds.length;

		// do a quick error check for single node cycle
		for (int i = 0; i < depLength; ++i) {
			if (messageId == dependencyIds[i]) {
				graphError = true;
				throw new NetworkException("Node cycle detected in message dependency graph");
			}
		}

		// add the message node
		// it might already exist, which is fine
		Message message = addMessage(messageId, currentTimeMs);
		if ((message.flags & Message.FLAG_ADDED) != 0) {
			// this message has already been added explicitly, we shouldn't be adding it twice
			graphError = true;
			throw new NetworkException("Duplicate message specified in message dependency graph");
		}
		message.flags |= Message.FLAG_ADDED;

		// add each dependency node
		for (int i = 0; i < depLength; ++i) {
			// if the node already exists, the existing one will be returned
			Message depMessage = addMessage(dependencyIds[i], currentTimeMs);
			// only add the dependency if the node isn't already executed
			if ((depMessage.flags & Message.FLAG_EXECUTED) == 0) {
				++message.predecessorCount;
				// add to front of linked list
				Dependency dep = new Dependency();
				dep.messageId = messageId;
				dep.next = depMessage.successorList;
				depMessage.successorList = dep;
			}
		}
	}

	// internal method to actually add a message node to the message map
	// dependencies must be created after this is called
	private Message addMessage(int messageId, long currentTimeMs) {
		Message message = messages.get(messageId);
		// if the message already exists, simply add it
		if (message != null) {
			return message;
		}

		// create the message and add an entry in the expiration queue
		ExpirationTime et = new ExpirationTime();
		et.addTimeMs = currentTimeMs;
		et.messageId = messageId;
		messageExpirationQueue.add(et);

		message = new Message();
		message.flags = 0; // initially message is not added, executed. or ready for execution
		// create default predecessors/successors here
		message.predecessorCount = 0;
		message.successorList = null;
		messages.put(messageId, message);
		return message;
	}

	/**
	 * Returns whether the message with the given ID has been executed.
	 *
	 * @param   messageId the ID of the message to check.
	 * @return  {@code true} if the message with the given ID exists and has been executed and {@code false} otherwise.
	 */
	public final boolean hasMessageExecuted(int messageId) {
		Message message = messages.get(messageId);
		return (message != null && (message.flags & Message.FLAG_EXECUTED) != 0);
	}

	/**
	 * Marks the message with the given ID as "ready to execute". If the message has no dependencies, it is immediately
	 * executed and its children are recursively examined in this way.
	 *
	 * @param   messageId the ID of the message to mark as "ready to execute".
	 * @throws  NetworkException if the graph has been previously invalidated, the message does not exist, or the
	 *          message has already been executed. If cycles exist, they will be detected here.
	 */
	public final void setMessageReadyToExecute(int messageId)
			throws NetworkException {
		if (graphError) {
			throw new NetworkException("Message dependency graph is invalid");
		}

		Message message = messages.get(messageId);
		if (message == null) {
			graphError = true;
			throw new NetworkException("Attempted to execute nonexistent message in message dependency graph");
		}

		message.flags |= Message.FLAG_READY_TO_EXECUTE;
		tryExecuteMessage(messageId);
	}

	/**
	 * Expires old messages. This method should be called at a regular interval. If an unexecuted message expires,
	 * an error is triggered.
	 *
	 * @param   currentTimeMs the current time in milliseconds.
	 * @throws  NetworkException if the graph has been previously invalidated or if an unexecuted message expires.
	 */
	public final void expireMessages(long currentTimeMs)
			throws NetworkException {
		if (graphError) {
			throw new NetworkException("Message dependency graph is invalid");
		}

		while (!messageExpirationQueue.isEmpty()) {
			if (messageExpirationQueue.peek().addTimeMs + messageExpirationTimeMs >= currentTimeMs) {
				// message isn't old enough to be removed, meaning no messages after it will be either
				return;
			}

			ExpirationTime et = messageExpirationQueue.remove();
			if ((messages.remove(et.messageId).flags & Message.FLAG_EXECUTED) == 0) {
				// the message was never executed
				// this means we're not receiving guaranteed messages
				graphError = true;
				throw new NetworkException("Unexecuted message expired in message dependency graph");
			}
		}
	}

	// attempts to execute a message if it has no incoming edges
	// if successful, outgoing edges from this message are removed and this method is called recursively on children
	private void tryExecuteMessage(int messageId)
			throws NetworkException {
		if (graphError) {
			throw new NetworkException("Message dependency graph is invalid");
		}

		// check to see if all dependencies have been completed
		Message message = messages.get(messageId);
		// make sure the message is ready to execute
		if ((message.flags & Message.FLAG_READY_TO_EXECUTE) == 0) {
			return;
		} if ((message.flags & Message.FLAG_EXECUTED) != 0) {
			// the message has already been executed - something is wrong with the graph
			graphError = true;
			throw new NetworkException("Attempted to execute a message that was already executed in message dependency graph");
		}

		// if there are no dependencies, this message is ready to be executed
		if (message.predecessorCount == 0) {
			// execute the message
			executeMessage(messageId);
			message.flags |= Message.FLAG_EXECUTED;
			// try executing all messages dependent on this one
			while (message.successorList != null) {
				// since this message is now complete, we can remove dependencies on it
				Dependency successor = message.successorList;
				message.successorList = successor.next;
				--messages.get(successor.messageId).predecessorCount;
				tryExecuteMessage(successor.messageId);
			}
		}
	}

	// if this flag is set to true, there is an error in the graph structure
	private boolean graphError;

	// used to form a linked list of dependencies
	private static class Dependency {
		public int messageId;
		public Dependency next;
	}

	// represents a message
	private static class Message {
		// indicates message has been explicitly added (as opposed to being created implicitly by adding another message)
		public static final int FLAG_ADDED              = 0x1;

		// indicates message is ready to execute (when no remaining dependencies)
		public static final int FLAG_READY_TO_EXECUTE   = 0x2;

		// indicates message has been execute
		public static final int FLAG_EXECUTED           = 0x4;

		public int flags;                   // combination of the above flags
		public int predecessorCount;        // number of incoming edges (dependencies of this node)
		public Dependency successorList;    // list of outgoing edges (dependencies of other nodes)

		// special case: 0-length successor list is treated as null to avoid unnecessary allocations
	}
	private HashMap<Integer, Message> messages; // map of messages

	// represents a time for a message to expire
	private static class ExpirationTime {
		public long addTimeMs;  // time the message was added
		public int messageId;   // ID of the message
	}
	private Queue<ExpirationTime> messageExpirationQueue;   // queue of message to process for expiration
	private int messageExpirationTimeMs;                    // time in milliseconds for messages to expire
}
